package com.example.rocketmq.service;

import com.alibaba.csp.sentinel.Entry;
import com.alibaba.csp.sentinel.SphU;
import com.alibaba.csp.sentinel.slots.block.BlockException;
import com.example.rocketmq.config.RocketMQProperties;
import com.example.rocketmq.model.Message;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.time.LocalDateTime;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 基于 Sentinel 的限流消费者服务
 * 使用 Sentinel 匀速器模式实现消息的匀速消费
 * 
 * @author demo
 */
@Slf4j
@Service
public class SentinelRateLimitConsumerService {

    @Autowired
    private RocketMQProperties properties;

    private DefaultMQPushConsumer consumer;
    
    // 统计信息
    private final AtomicInteger consumedThisMinute = new AtomicInteger(0);
    private final AtomicLong totalConsumed = new AtomicLong(0);
    private final AtomicLong totalBlocked = new AtomicLong(0);
    private final AtomicLong lastResetTime = new AtomicLong(System.currentTimeMillis());

    @PostConstruct
    public void init() throws Exception {
        consumer = new DefaultMQPushConsumer(properties.getConsumer().getGroup());
        consumer.setNamesrvAddr(properties.getNamesrvAddr());
        consumer.subscribe(properties.getTopic().getName(), "*");
        
        // 设置消息监听器
        consumer.setMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    try {
                        // 使用 Sentinel 进行限流控制
                        try (Entry entry = SphU.entry("messageConsume")) {
                            // 通过限流检查，可以消费消息
                            processMessage(msg);
                            consumedThisMinute.incrementAndGet();
                            totalConsumed.incrementAndGet();
                            
                            log.info("✅ 消息消费成功，消息ID: {}, 当前分钟已消费: {}/{}, 总计消费: {}", 
                                    msg.getKeys(), consumedThisMinute.get(), 
                                    properties.getConsumer().getRateLimit().getMessagesPerMinute(),
                                    totalConsumed.get());
                            
                        } catch (BlockException e) {
                            // 被限流，消息将被排队等待
                            totalBlocked.incrementAndGet();
                            log.warn(" 消息被限流，消息ID: {}, 当前分钟已消费: {}/{}, 总计被限流: {}",
                                    msg.getKeys(), consumedThisMinute.get(), 
                                    properties.getConsumer().getRateLimit().getMessagesPerMinute(),
                                    totalBlocked.get());
                            
                            // 返回稍后重试，消息不会被丢弃
                            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                        }
                        
                    } catch (Exception e) {
                        log.error("❌ 消费消息失败，消息ID: {}", msg.getKeys(), e);
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        
        consumer.start();
        log.info("� Sentinel 限流消费者启动成功，NameServer: {}, 消费者组: {}, 限流配置: 每分钟{}条消息", 
                properties.getNamesrvAddr(), properties.getConsumer().getGroup(),
                properties.getConsumer().getRateLimit().getMessagesPerMinute());
    }

    @PreDestroy
    public void destroy() {
        if (consumer != null) {
            consumer.shutdown();
            log.info("� Sentinel 限流消费者已关闭");
        }
    }

    /**
     * 处理消息
     */
    private void processMessage(MessageExt msg) {
        String messageBody = new String(msg.getBody());
        log.info("� 正在消费消息: Topic={}, Tags={}, Keys={}, Body={}", 
                msg.getTopic(), msg.getTags(), msg.getKeys(), messageBody);
        
        // 模拟业务处理
        try {
            // 模拟处理时间
            Thread.sleep(100);
            log.info("✅ 消息处理完成: {}", msg.getKeys());
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("❌ 消息处理被中断: {}", msg.getKeys(), e);
        }
    }

    /**
     * 每分钟重置限流计数器
     */
    public void resetRateLimitCounter() {
        consumedThisMinute.set(0);
        log.info("� 重置限流计数器，新的一分钟开始");
    }

    /**
     * 获取当前消费统计信息
     */
    public String getConsumptionStats() {
        long currentTime = System.currentTimeMillis();
        long lastReset = lastResetTime.get();
        long remainingTime = Math.max(0, 60000 - (currentTime - lastReset));
        
        return String.format("� Sentinel 限流统计 - 当前分钟已消费: %d/%d, 总计消费: %d, 总计被限流: %d, 下次重置剩余时间: %d秒", 
                consumedThisMinute.get(), 
                properties.getConsumer().getRateLimit().getMessagesPerMinute(),
                totalConsumed.get(),
                totalBlocked.get(),
                remainingTime / 1000);
    }

    /**
     * 获取当前分钟已消费消息数
     */
    public int getConsumedThisMinute() {
        return consumedThisMinute.get();
    }

    /**
     * 获取总计消费消息数
     */
    public long getTotalConsumed() {
        return totalConsumed.get();
    }

    /**
     * 获取总计被限流消息数
     */
    public long getTotalBlocked() {
        return totalBlocked.get();
    }
}
